
import { fork } from "child_process";
import EventEmitter from "events";
import { user, sendReponse, sleep } from "./declare";

// 批量制造用户
const getUsers = (count = 100 * 1000): user[] => Array.from(new Array(count)).map((_, i) => ({ id: i + 1, name: "user" + (i + 1) }))


function createSender(Id: number /*, signal: AbortSignal*/) {

    const sender = fork('./send'/*, { signal }*/)

    var emiter = new EventEmitter()
    //emiter.setMaxListeners(20)

    sender.on('message', (data: any) => emiter.emit('success', data))
    sender.on('error', (err: any) => emiter.emit('fail', err))

    return {
        _sender: sender,
        sendId: Id,
        _status: 0,
        get status() {
            return this._status
        },
        lock: function () {
            this._status = 1
        },
        release: function () {
            this._status = 0
        },
        sendData: (users: user[]) => {
            sender.send({ type: 'normal', data: users })

            emiter.removeAllListeners('success')
            emiter.removeAllListeners('fail')

            return new Promise((resolve, reject) => {
                emiter.once('success', (data: any) => {
                    resolve(data)
                })
                emiter.once('fail', (err: any) => {
                    reject(err)
                })
            })
        },
        close() {
            sender.send({ type: 'exit', data: null })
        }
    }

}

async function Start() {
    const tempUsers = getUsers(8 * 1000)
    const chunkSize = 30;
    let max = 10
    //const sendResult: sendReponse[] = []

    // node15 新属性
    // const controller = new AbortController();
    // const { signal } = controller;

    const senders: any[] = []
    for (let index = 0; index < max; index++) {
        senders.push(createSender(index/*,signal */))
    }
    const getIdleSender = () => {
        const sender = senders.find(e => e.status === 0)
        if (sender) {
            sender.lock()
        }
        return sender
    }
    const getOccupySender = () => {
        return senders.filter(e => e.status === 1)
    }

    let currentPointer = 0
    while (currentPointer <= tempUsers.length) {
        const currentSender = getIdleSender();
        if (currentSender) {
            console.log('当前任务', currentSender.sendId)

            currentSender.sendData(tempUsers.slice(currentPointer, currentPointer += chunkSize)).then((e: any) => {

                currentSender.release()
                console.log('~~释放任务', currentSender.sendId)
                //sendResult.push(e)
            })
            continue
        }
        await sleep(100)
    }

    //等待剩余任务完成
    while (getOccupySender().length != 0) {
        await sleep(100)
    }

    for (const item of senders) {
        item.close()
    }
    console.log('所有任务已结束')
}

Start()